package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

/**
 * Author: Felix
 * Date: 2021/8/4
 * Desc: 自定义反序列化器
 */
public class MyDeserializationSchemaFunction01 implements DebeziumDeserializationSchema<String> {
    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //获取Struct对象
        Struct valueStruct = (Struct) sourceRecord.value();
        Struct sourceStrut = valueStruct.getStruct("source");
        //获取数据库的名称
        String database = sourceStrut.getString("db");
        //获取表名
        String table = sourceStrut.getString("table");

        //获取类型
        //Envelope对增删改查做了封装   toLowerCase转换小写
        String type = Envelope.operationFor(sourceRecord).toString().toLowerCase();
        if(type.equals("create")){
            type="insert";
        }

        //补充字段加入进去
        JSONObject jsonObj = new JSONObject();
        jsonObj.put("database",database);
        jsonObj.put("table",table);
        jsonObj.put("type",type);

        //获取影响的数据data
        // 源格式：id=1,name=liuxintong,age=17
        //目标格式：{"id":74603,"order_id":28641,"order_status":"1005","operate_time":"2021-07-30 11:35:49"}}
        Struct afterStruct = valueStruct.getStruct("after");
        JSONObject dataJsonObj = new JSONObject();
        //删除的时候为空会报错
        if(afterStruct!=null){
            //获取所有的数据
            for (Field field : afterStruct.schema().fields()) {
                //属性名称
                String fieldName = field.name();
                //属性名称属性值
                Object fieldValue = afterStruct.get(field);
                //把属性名和属性值转成json
                dataJsonObj.put(fieldName,fieldValue);
            }
        }
        //把数据加入进去
        jsonObj.put("data",dataJsonObj);

        //向下游传递数据
        collector.collect(jsonObj.toJSONString());
    }
    @Override
    public TypeInformation<String> getProducedType() {
        //返回的是他自己
        return TypeInformation.of(String.class);
    }
}
